package org.dh2580.delay.queue.bean.task;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.dh2580.delay.queue.bean.bean.DelayQueueBean;
import org.dh2580.delay.queue.bean.db.DBUtil;
import org.dh2580.delay.queue.bean.model.DelayMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author: hao.deng
 * @Date: 2019/5/15-下午2:31
 */
public class ClearMsgTask implements Runnable {

    private static Logger LOG = LoggerFactory.getLogger(ClearMsgTask.class);

    private static final String QUERY_ABANDON_MSG_TPL = "SELECT id AS id, route_key AS routeKey, msg AS msg, ttl AS ttl, ack_state AS ackState, create_time AS createTime FROM tb_delay_msg_queue WHERE ttl < ? AND ack_state = 0;";

    private static final String BATCH_ABANDON_MSG_TPL = "UPDATE tb_delay_msg_queue SET ack_state = 3 WHERE id IN ( ? );";

    @Override
    public void run() {
        try {
            long ttl = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(DelayQueueBean.ABANDON_THRESHOLD_DAY);

            List<DelayMsg> delayMsgs = DBUtil.query(QUERY_ABANDON_MSG_TPL, DelayMsg.class, ttl);
            if (delayMsgs == null || delayMsgs.size() == 0) {
                return;
            }

            List<Long> msgIds = new ArrayList<>(delayMsgs.size());
            for (DelayMsg delayMsg : delayMsgs) {
                msgIds.add(delayMsg.getId());
            }

            StringBuilder idsStr = new StringBuilder();

            for (Long msgId : msgIds) {
                idsStr.append(msgId);
                idsStr.append(",");
            }

            DBUtil.execute(BATCH_ABANDON_MSG_TPL.replace("?", idsStr.subSequence(0, idsStr.length() - 1)));

        } catch (SQLException e) {
            LOG.error("Execute Sql Except", e);
        } catch (Exception e) {
            LOG.error("Execute Except", e);
        }
    }
}
